#include "IOQueue.h"

using namespace Hiranipra;
using namespace Hiranipra::Platform;
using namespace Hiranipra::Platform::IO;

class IOQueue::IORequest : implements RefInstance
{
public:
	enum RequestType
	{
		ReadRequest,
		WriteRequest,
	};
public:
	uint32			RequestID;
	RequestType		Type;
	bool			IsCancelled;
	bool			IsComplete;
	IFileStream*	Stream;
	MemoryBuffer*	Buffer;

	IORequest( uint32 requestId, RequestType type, IFileStream* stream, MemoryBuffer* buffer ) :
		RequestID( requestId ), Type( type ), IsCancelled( false ), IsComplete( false ),
		Stream( stream ), Buffer( buffer )
	{
		stream->Retain();
		buffer->Retain();
	}

	~IORequest()
	{
		RELEASE( this->Stream );
		RELEASE( this->Buffer );
	}
};

class IOQueue::IORunner : public Runnable
{
public:
	IOQueue*	Queue;
	bool		Stop;

	IORunner( IOQueue* queue ) :
		Queue( queue ), Stop( false )
	{
	}

	void Abort( Thread* thread )
	{
		this->Stop = true;
		thread->interrupt();
		thread->wait();
	}

	void run()
	{
		while( this->Stop == false )
		{
			IORequest* request;
			try
			{
				request = Queue->_requests->next();
			}
			catch( Interrupted_Exception& )
			{
				Queue->_requests->cancel();
				break;
			}
			if( request->IsCancelled == true )
			{
				RELEASE( request );
				continue;
			}

			IOEvent::IOEventName eventName;
			bool succeeded = false;
			switch( request->Type )
			{
			default:
			case IORequest::ReadRequest:
				eventName = IOEvent::ReadCompleted;
				succeeded = ( request->Stream->Read( request->Buffer ) > 0 );
				break;
			case IORequest::WriteRequest:
				eventName = IOEvent::WriteCompleted;
				succeeded = ( request->Stream->Write( request->Buffer ) > 0 );
				break;
			}

			request->IsComplete = true;

			IOEvent ev( eventName, request->RequestID, succeeded, request->Stream, request->Buffer );
			Queue->EventSource.Fire( ev );

			RELEASE( request );
		}
	}
};

IOQueue::IOQueue( IFileSystem* fileSystem ) :
	_fs( fileSystem )
{
	fileSystem->Retain();
	_requests = new BlockingQueue<IORequest*, Mutex>();

	_ioRunner = new IORunner( this );
	Task ioRunnerTask( _ioRunner );
	_ioThread = new Thread( ioRunnerTask );
}

IOQueue::~IOQueue()
{
	_ioRunner->Abort( _ioThread );
	DELETE_PTR( _ioThread );

	RELEASE( _fs );
	DELETE_PTR( _requests );
}

uint32 IOQueue::BeginRead( IFileStream* sourceStream, MemoryBuffer* buffer )
{
	uint32 requestId = _requestIds++;
	IORequest* request = new IORequest( requestId, IORequest::ReadRequest, sourceStream, buffer );
	_requests->add( request );
	return requestId;
}

uint32 IOQueue::BeginWrite( IFileStream* targetStream, MemoryBuffer* buffer )
{
	uint32 requestId = _requestIds++;
	IORequest* request = new IORequest( requestId, IORequest::WriteRequest, targetStream, buffer );
	_requests->add( request );
	return requestId;
}

bool IOQueue::IsComplete( uint32 requestId ) const
{
	return false;
}

bool IOQueue::Cancel( uint32 requestId )
{
	return false;
}
